package com.shujia.jinjie;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

public class ShuJiaInterceptor implements Interceptor {

    @Override
    public void initialize() {
        // 这个方法主要是初始化一些资源，比如连接数据库，文件，网络等
    }

    @Override
    public Event intercept(Event event) {
        //每一个event的处理方案
        // 这个方法很重要，针对source将一条数据封装的event进行拦截，对event进行修改，返回一个修改后的event
        //event包括了header和body，其中body是byte[]类型，需要转换为string，然后进行修改，再转换为byte[]，返回一个修改后的event
        //需求: 如果event中的body中包含"shujia"，则修改header中的key为"type", value为"sj"；反之将key为"type", value为"nsj"
        String info = new String(event.getBody());
        if (info.contains("shujia")) {
            event.getHeaders().put("type", "sj");
        } else {
            event.getHeaders().put("type", "nsj");
        }

        return event;

    }

    @Override
    public List<Event> intercept(List<Event> list) {
        ArrayList<Event> events = new ArrayList<>();

        for (Event event : list) {
            Event event1 = intercept(event);
            events.add(event1);
        }

        return events;
    }

    @Override
    public void close() {
        // 这个方法主要是释放initialize()中创建的资源，比如连接数据库，文件，网络等
    }

    public static class Builder implements Interceptor.Builder {
        @Override
        public Interceptor build() {
            return new ShuJiaInterceptor();
        }


        @Override
        public void configure(Context context) {
            // 这个方法主要是对context进行配置，比如获取配置文件中的参数，进行一些初始化操作
        }
    }

}
